package com.bbx.flink.dataStream_api_all.CDC;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDCDemo {

    public static void main(String [] args) throws Exception {

        //执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取source
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("192.168.10.7")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("maxwell_test1", "maxwell_test2")
                .tableList("maxwell_test1.test1", "maxwell_test1.test2", "maxwell_test2.test3")
//                .deserializer(new StringDebeziumDeserializationSchema())
                .deserializer(new CustomerDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        env.addSource(sourceFunction)    //设置source
                .setParallelism(1)     //设置并行度
                .print();              //输出

        //执行
        env.execute("print CDC");
    }





}
